package cn.lesaas.esb.poc.test;

import org.apache.camel.CamelContext;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.sql.SqlComponent;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.impl.JavaUuidGenerator;
import org.apache.camel.impl.SimpleRegistry;
import org.apache.camel.spring.spi.SpringTransactionPolicy;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;

import com.mysql.jdbc.jdbc2.optional.MysqlDataSource;

import cn.lesaas.esb.poc.mock.VBSRoute;
import cn.lesaas.esb.poc.route.JobProcess;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

public class ScheduleRoute extends RouteBuilder {

	@Override
	public void configure() throws Exception {
		// 轮训任务表
		from("scheduler://foo?delay=1000").startupOrder(30).to("direct:sql");

		from("direct:sql").transacted().setHeader("sequence").method(JavaUuidGenerator.class, "generateUuid")
				.to("sql:update jobs set state=1, sequence=:#sequence where state=0 and UNIX_TIMESTAMP() - UNIX_TIMESTAMP(startTimestamp)  > duration")
				.to("sql:select message from jobs where sequence=:#sequence").to("direct:Parallel");

		from("direct:Parallel").split(body()).parallelProcessing()
				.log("Processing message[${property.CamelSplitIndex}]")
				.to("log:all?showAll=true")
				.setBody(simple("${body[message]}"))
				.log("[${body}]")
				.to("netty4:tcp://172.16.100.41:59999?allowDefaultCodec=false&encoder=#stringEncoder&decoder=#stringDecoder").end();
	}

	public static void main(String args[]) throws Exception {
		SimpleRegistry registry = new SimpleRegistry();
		MysqlDataSource dataSource = new MysqlDataSource();
        dataSource.setURL("jdbc:mysql://127.0.0.1:3306/poc?user=root&password=mysql");
        
		DataSourceTransactionManager transactionManager = new DataSourceTransactionManager(dataSource);
		registry.put("transactionManager", transactionManager);
		SpringTransactionPolicy propagationRequired = new SpringTransactionPolicy();
		propagationRequired.setTransactionManager(transactionManager);
		propagationRequired.setPropagationBehaviorName("PROPAGATION_REQUIRED");
		registry.put("PROPAGATION_REQUIRED", propagationRequired);
		registry.put("stringEncoder", new StringEncoder());
		registry.put("stringDecoder", new StringDecoder());
		CamelContext context = new DefaultCamelContext(registry);

		SqlComponent sqlComponent = new SqlComponent();
		sqlComponent.setDataSource(dataSource);
		context.addComponent("sql", sqlComponent);
		
		context.addRoutes(new VBSRoute());
		context.addRoutes(new ScheduleRoute());

		// start the route and let it do its work
		context.start();
		Thread.sleep(100000);

		// stop the CamelContext
		context.stop();
	}
}
